package com.aotain.nyx.areastatis;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map.Entry;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

import com.aotain.apollo.mongodb.MongoImportTool;
import com.aotain.common.mongo.DataColumn;
import com.aotain.nyx.ipstat.IPStatTuple;
import com.aotain.nyx.statis.AbnStatisTuple;

public class AreaStatisImport implements FlatMapFunction<Tuple2<String,AbnStatisTuple>	, Object> {

	/** 
	 * @Fields serialVersionUID : TODO(用一句话描述这个变量表示什么) 
	 */ 
	private static final long serialVersionUID = 6706009868030895225L;

	@Override
	public void flatMap(Tuple2<String, AbnStatisTuple> value,
			Collector<Object> out) throws Exception {
		// TODO Auto-generated method stub
		String tableName = "SDS_ABNORMAL_AREA";
		String dip = value.f1.getDestIP();
		AbnStatisTuple col = value.f1;
		if(col == null)
			return;

		List<DataColumn> saverow = new ArrayList<DataColumn>();
		DataColumn colDIP = new DataColumn();
		colDIP.setKey("IP");
		colDIP.setValue(dip);
		saverow.add(colDIP);
		DataColumn colProvince = new DataColumn();
		colProvince.setKey("AREA");
		colProvince.setValue(col.getSourceProvince());
		saverow.add(colProvince);
		DataColumn colCountry = new DataColumn();
		colCountry.setKey("COUNTRY");
		colCountry.setValue(col.getSourceCountry());
		saverow.add(colCountry);

		SimpleDateFormat sdfd = new SimpleDateFormat("yyyyMMdd");
		Date date = new Date();
		String dateDay = sdfd.format(date);
		String reporttime = dateDay;

		DataColumn colReportTime = new DataColumn();
		colReportTime.setKey("REPORTTIME");
		colReportTime.setValue(Long.parseLong(reporttime));
		saverow.add(colReportTime);

		List<DataColumn> incrow = new ArrayList<DataColumn>();
		DataColumn colFlow = new DataColumn();
		colFlow.setKey("PV");
		colFlow.setValue(col.getUpStreamPacket());
		incrow.add(colFlow);
		
		DataColumn colAttNum = new DataColumn();
		colAttNum.setKey("ATTACKNUM");
		colAttNum.setValue(col.getAttackNum());
		incrow.add(colAttNum);
		
		DataColumn colStAttNum = new DataColumn();
		colStAttNum.setKey("STPROVINCEPV");
		colStAttNum.setValue(col.getStAttackNum());
		incrow.add(colStAttNum);

		List<DataColumn> filter = new ArrayList<DataColumn>();
		filter.add(new DataColumn("REPORTTIME",Long.parseLong(reporttime)));
		filter.add(new DataColumn("IP",dip));
		filter.add(new DataColumn("AREA",col.getSourceProvince()));


		MongoImportTool.getInstance().UpdateRowData(tableName, filter, saverow);

		MongoImportTool.getInstance().IncRowData(tableName, filter, incrow);
	}

}
